package com.lg.monitor

import java.sql.{Connection, DriverManager}

import com.lg.bean.BusInfo
import org.apache.spark.sql.ForeachWriter

/**
 * @Author Cao Dun
 * @Date 2021/8/1 14:12
 * @Version 1.0
 * @Description
 */

class MysqlWriter extends  ForeachWriter[BusInfo] {

  private var connection: Connection = null;

  override def open(partitionId: Long, epochId: Long): Boolean = {
    Class.forName("com.mysql.jdbc.Driver")
    connection = DriverManager.getConnection("jdbc:mysql://linux123:3306/lg_bus_info?serverTimezone=UTC&characterEncoding=UTF-8&useSSL=false", "root", "12345678")
    return true
  }


  override def process(value: BusInfo): Unit = {
    val stmt = connection.createStatement()
    //车牌号
    var deployNum = value.deployNum
    //经纬度信息
    val strings: Array[String] = value.lglat.split("_")
    val longitude = strings(0)
    val latitude = strings(1)
    println(deployNum + " " + longitude + " " + latitude);
    /**
     * INSERT INTO bus_info (carId,longitude,latitude) VALUES ('xx','11','22');
     */
    stmt.execute(s"""INSERT INTO bus_info (deployNum,longitude,latitude) VALUES ('$deployNum','$longitude','$latitude')""")
    //插入mysql表
    println(deployNum + " " + longitude + " " + latitude);
  }

  override def close(errorOrNull: Throwable): Unit =
    connection.close()
}
